package net.self.kafka.consumer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import java.util.Properties;


public class SelfSignedKafkaProducer {

    public static Producer<String, String> createProducer() {
        Properties props = new Properties();
        String brokerList = System.getenv("BROKER_LIST");
        if ((brokerList) == null){
            props.put("bootstrap.servers", KafkaProducerConfig.BOOTSTRAP_SERVERS);
        } else {
            props.put("bootstrap.servers", brokerList);
        }
        props.put("acks", KafkaProducerConfig.KAFKA_REQUEST_REQUIRED_ACKS);
        props.put("timeout.ms", KafkaProducerConfig.TIMEOUT_MS);
        props.put("request.timeout.ms", KafkaProducerConfig.REQUEST_TIMEOUT_MS);
        props.put("reconnect.backoff.ms", KafkaProducerConfig.RECONNECT_BACKOFF_MS);
        props.put("retry.backoff.ms", KafkaProducerConfig.RETRY_BACKOFF_MS);
        props.put("retries", KafkaProducerConfig.KAFKA_RETRIES);
        props.put("batch.size", KafkaProducerConfig.BATCH_SIZE);
        props.put("linger.ms", KafkaProducerConfig.LINGER_MS);
        props.put("buffer.memory", KafkaProducerConfig.BUFFER_MEMORY);
        props.put("key.serializer", KafkaProducerConfig.KEY_SERIALIZER);
        props.put("value.serializer", KafkaProducerConfig.VALUE_SERIALIZER);

        return new KafkaProducer(props);
    }

}
